/**
 * Copyright (c) 2013-2019 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.pubsub;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;

import org.redisson.PubSubEntry;
import org.redisson.api.RFuture;
import org.redisson.client.BaseRedisPubSubListener;
import org.redisson.client.ChannelName;
import org.redisson.client.RedisPubSubListener;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.misc.RPromise;
import org.redisson.misc.RedissonPromise;
import org.redisson.misc.TransferListener;

/**
 * 抽象的PubSub类
 */
abstract class PublishSubscribe<E extends PubSubEntry<E>> {
    // final类型的PublishSubscribeService属性。在当前类或子类的实例上引用地址不可变。
    // 然而由于每一个锁实例实际上都是使用相同引用地址的PublishSubscribe，因此这里的service也是相同引用地址
    // 即所有锁实例使用相同的PublishSubscribeService
    private final PublishSubscribeService service;

    PublishSubscribe(PublishSubscribeService service) {
        super();
        this.service = service;
    }

    // 一个保存当前服务器上所有RedissonLockEntry(即所有订阅通道)的Map
    // 还是上面的原因，由于每一个锁实例使用相同引用地址的PublishSubscribe，因此entries也是相同引用地址。也就是所有的锁实例都是在同一个Map中增删Entry
    // 该Map将使用Lock的EntryName作为key，非公平锁下EntryName的细粒度是锁名，公平锁下EntryName的细粒度是线程。
    private final ConcurrentMap<String, E> entries = new ConcurrentHashMap<>();

    public void unsubscribe(E entry, String entryName, String channelName) {
        AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
        semaphore.acquire(new Runnable() {
            @Override
            public void run() {
                if (entry.release() == 0) {
                    // just an assertion
                    boolean removed = entries.remove(entryName) == entry;
                    if (!removed) {
                        throw new IllegalStateException();
                    }
                    service.unsubscribe(new ChannelName(channelName), semaphore);
                } else {
                    semaphore.release();
                }
            }
        });

    }

    /**
     * 线程获取锁失败，调用该方法订阅通道监听锁释放的消息。
     * 所有的锁实例会操作相同引用地址的方法，向entries中增删各自的Entry。因此entries中会存在各种不同名称形式的EntryName
     *
     * @param entryName   看情况。不同类型锁的EntryName也不相同。重复非公平下 ConnectionId:lockName
     * @param channelName 看情况。不同类型锁订阅的channelName也不相同。重入非公平下 redisson_lock__channel:{lockName}
     * @return 固定为 RFuture<RedissonLockEntry>
     */
    public RFuture<E> subscribe(String entryName, String channelName) {
        // new一个异步任务--原子引用类。 listenerHolder.get() 就是异步任务本体
        AtomicReference<Runnable> listenerHolder = new AtomicReference<Runnable>();

        // 根据channelName获取一个AsyncSemaphore。
        // 非公平锁时，同一把锁的所有阻塞线程都将订阅相同的channelName。而公平锁时，同一把锁的所有线程订阅各自的channelName
        // AsyncSemaphore用于限制同时向redis发送SUB命令的线程数量
        AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));

        /**
         * 下面设置的两个异步任务，实际执行是在 semaphore.acquire(listener) 这句中
         */

        RPromise<E> newPromise = new RedissonPromise<E>() {
            @Override
            public boolean cancel(boolean mayInterruptIfRunning) {
                return semaphore.remove(listenerHolder.get());
            }
        };

        Runnable listener = new Runnable() {
            @Override
            public void run() {
                // 查看当前服务器是否已存在相同EntryName的RedissonLockEntry
                E entry = entries.get(entryName);

                // 如果有，则直接复用
                if (entry != null) {
                    // 线程数+1
                    entry.acquire();
                    // 释放AsyncSemaphore信号量
                    semaphore.release();
                    // 确保真正从AsyncSemaphore中删除
                    entry.getPromise().onComplete(new TransferListener<E>(newPromise));
                    return;
                }

                // 如果没有，则创建一个RedissonLockEntry. 这里调用的是LockPubSub中的createEntry()方法
                E value = createEntry(newPromise);
                // 线程数+1
                value.acquire();

                // 使用putIfAbsent，防止在此时突然有其他线程插入相同entryName的RedissonLockEntry
                E oldValue = entries.putIfAbsent(entryName, value);
                // 如果oldValue非空，说明被人插队了，那就还是复用呗，和上面相同的逻辑
                if (oldValue != null) {
                    oldValue.acquire();
                    semaphore.release();
                    oldValue.getPromise().onComplete(new TransferListener<E>(newPromise));
                    return;
                }

                // 创建订阅监听器。监听redis发送的锁释放通知。见下方createListener
                // 注意value是RedissonLockEntry，可能会含有多个
                RedisPubSubListener<Object> listener = createListener(channelName, value);

                // 真正向redis发送命令，订阅指定通道的方法
                service.subscribe(LongCodec.INSTANCE, channelName, semaphore, listener);
            }
        };

        // 用AsyncSemaphore去acquire上面的listener。
        // 这样的话订阅一个锁并不是只要获取锁失败就去订阅，而是必须在上一个获取失败的线程订阅完后才会执行
        semaphore.acquire(listener);
        listenerHolder.set(listener);

        return newPromise;
    }

    protected abstract E createEntry(RPromise<E> newPromise);

    protected abstract void onMessage(E value, Long message);

    /**
     * 创建监听器，监听redis的锁释放通知。
     * @param channelName 订阅的通道名称
     * @param value
     * @return
     */
    private RedisPubSubListener<Object> createListener(String channelName, E value) {
        RedisPubSubListener<Object> listener = new BaseRedisPubSubListener() {
            /**
             * 收到redis通知时要做的事情
             * @param channel redis向哪个通道发送的消息
             * @param message 消息
             */
            @Override
            public void onMessage(CharSequence channel, Object message) {
                // 首先判断redis发送的通道是不是当前线程订阅的通道
                if (!channelName.equals(channel.toString())) {
                    return;
                }

                // 执行this.中的onMessage()方法。本类中未实现，因此实际上是执行子类LockPubSub中的onMessage()方法
                PublishSubscribe.this.onMessage(value, (Long) message);
            }

            @Override
            public boolean onStatus(PubSubType type, CharSequence channel) {
                if (!channelName.equals(channel.toString())) {
                    return false;
                }

                if (type == PubSubType.SUBSCRIBE) {
                    value.getPromise().trySuccess(value);
                    return true;
                }
                return false;
            }

        };
        return listener;
    }

}
